/*-------------------------------------------------------------------------
 *
 * execAsync.c
 *	  支持异步执行的例程
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/executor/execAsync.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "executor/execAsync.h"
#include "executor/executor.h"
#include "executor/nodeAppend.h"
#include "executor/nodeForeignscan.h"

/*
 * 异步请求一个来自设计的支持异步的节点的元组。
 */
void ExecAsyncRequest(AsyncRequest *fc_areq)
{
	if (fc_areq->requestee->chgParam != NULL)	/* 有什么变化吗？ */
		ExecReScan(fc_areq->requestee);	/* 让 ReScan 处理这个 */

	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStartNode(fc_areq->requestee->instrument);

	switch (nodeTag(fc_areq->requestee))
	{
		case T_ForeignScanState:
			ExecAsyncForeignScanRequest(fc_areq);
			break;
		default:
			/* 如果节点不支持异步，调用者搞砸了。 */
			elog(ERROR, "unrecognized node type: %d",
				 (int) nodeTag(fc_areq->requestee));
	}

	ExecAsyncResponse(fc_areq);

	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStopNode(fc_areq->requestee->instrument,
					  TupIsNull(fc_areq->result) ? 0.0 : 1.0);
}

/*
 * 给异步节点一个机会来配置它希望等待的文件描述符事件。我们期望节点类型特定的回调进行如下形式的单次调用：
 *
 * AddWaitEventToSet(set, WL_SOCKET_READABLE, fd, NULL, areq);
 */
void ExecAsyncConfigureWait(AsyncRequest *fc_areq)
{
	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStartNode(fc_areq->requestee->instrument);

	switch (nodeTag(fc_areq->requestee))
	{
		case T_ForeignScanState:
			ExecAsyncForeignScanConfigureWait(fc_areq);
			break;
		default:
			/* 如果节点不支持异步，调用者搞砸了。 */
			elog(ERROR, "unrecognized node type: %d",
				 (int) nodeTag(fc_areq->requestee));
	}

	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStopNode(fc_areq->requestee->instrument, 0.0);
}

/*
 * 当发生相关事件时，调用异步节点的回调。
 */
void ExecAsyncNotify(AsyncRequest *fc_areq)
{
	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStartNode(fc_areq->requestee->instrument);

	switch (nodeTag(fc_areq->requestee))
	{
		case T_ForeignScanState:
			ExecAsyncForeignScanNotify(fc_areq);
			break;
		default:
			/* 如果节点不支持异步，调用者搞砸了。 */
			elog(ERROR, "unrecognized node type: %d",
				 (int) nodeTag(fc_areq->requestee));
	}

	ExecAsyncResponse(fc_areq);

	/* 必须提供我们自己的仪器支持 */
	if (fc_areq->requestee->instrument)
		InstrStopNode(fc_areq->requestee->instrument,
					  TupIsNull(fc_areq->result) ? 0.0 : 1.0);
}

/*
 * 当异步节点产生结果时，调用请求者的回调。
 */
void ExecAsyncResponse(AsyncRequest *fc_areq)
{
	switch (nodeTag(fc_areq->requestor))
	{
		case T_AppendState:
			ExecAsyncAppendResponse(fc_areq);
			break;
		default:
			/* 如果节点不支持异步，调用者搞砸了。 */
			elog(ERROR, "unrecognized node type: %d",
				 (int) nodeTag(fc_areq->requestor));
	}
}

/*
 * 请求节点应该调用此函数将元组传递给其请求者节点。请求节点可以从其 ExecAsyncRequest 或 ExecAsyncNotify 回调中调用此函数。
 */
void ExecAsyncRequestDone(AsyncRequest *fc_areq, TupleTableSlot *fc_result)
{
	fc_areq->request_complete = true;
	fc_areq->result = fc_result;
}

/*
 * 请求节点应该调用此函数以指示它在等待回调。请求节点可以从其 ExecAsyncRequest 或 ExecAsyncNotify 回调中调用此函数。
 */
void ExecAsyncRequestPending(AsyncRequest *fc_areq)
{
	fc_areq->callback_pending = true;
	fc_areq->request_complete = false;
	fc_areq->result = NULL;
}
